package com.orion.config;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.BatchErrorHandler;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.ErrorHandler;

import javax.annotation.Resource;

/**
 * @author Administrator
 * @date 2021/12/6
 */
@Configuration
@EnableKafka
public class KafkaContainerConfig {

    @Value("${spring.kafka.listener.poll-timeout}")
    private Integer pollTimeout;

    @Value("${spring.kafka.listener.concurrency}")
    private Integer concurrency;

    @Value("${spring.kafka.listener.ack-mode}")
    private ContainerProperties.AckMode ackMode;

    @Value("${spring.kafka.listener.type}")
    private KafkaProperties.Listener.Type type;

    @Resource
    private ConsumerFactory<String, Object> consumerFactory;

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String>
                factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 设置消费者工厂
        factory.setConsumerFactory(consumerFactory);
        // 消费者组中线程数量
        factory.setConcurrency(concurrency);

        ContainerProperties containerProperties = factory.getContainerProperties();
        // 拉取超时时间
        containerProperties.setPollTimeout(pollTimeout);
        containerProperties.setAckMode(ackMode);
        if (type.equals(KafkaProperties.Listener.Type.BATCH)) {
            factory.setBatchListener(true);
            factory.setBatchErrorHandler(batchErrorHandler());
        } else {
            factory.setBatchListener(false);
            factory.setErrorHandler(errorHandler());
        }

        return factory;
    }

    @Bean
    @ConditionalOnProperty(name = "spring.kafka.listener.type", havingValue = "batch")
    public BatchErrorHandler batchErrorHandler() {
        System.out.println("init batchErrorHandler");
        BatchErrorHandler batchErrorHandler = new BatchErrorHandler(){
            @Override
            public void handle(Exception thrownException, ConsumerRecords<?, ?> data) {
                System.out.println("batch error handler , data : " + data + " , exception : " + thrownException);
            }
        };
        return batchErrorHandler;
    }


    @Bean
    @ConditionalOnMissingBean(BatchErrorHandler.class)
    public ErrorHandler errorHandler() {
        System.out.println("init errorHandler");
        /*return (ex, records, consumer) -> {
            MessageHeaders headers = records.get();
            List<String> topics = headers.get(KafkaHeaders.RECEIVED_TOPIC, List.class);
            List<Integer> partitions = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, List.class);
            List<Long> offsets = headers.get(KafkaHeaders.OFFSET, List.class);
            Map<TopicPartition, Long> offsetsToReset = new HashMap<>();
            for (int i = 0; i < topics.size(); i++) {
                int index = i;
                offsetsToReset.compute(new TopicPartition(topics.get(i), partitions.get(i)),
                        (k, v) -> v == null ? offsets.get(index) : Math.min(v, offsets.get(index)));
            }
            offsetsToReset.forEach((k, v) -> c.seek(k, v));
            return null;
        };*/
        return (thrownException, data) ->
                System.out.println("error handler , data : " + data + " , exception : " + thrownException);
    }
}
